package com.wan.net.rpc.invoker;

import cn.hutool.core.lang.Assert;
import com.wan.net.config.HostAndPort;
import com.wan.net.core.client.Client;
import com.wan.net.core.packet.Packet;
import com.wan.net.exception.RpcInvokeException;
import com.wan.net.rpc.RpcManager;
import com.wan.net.rpc.invoker.balancer.Balancer;
import com.wan.net.rpc.invoker.callback.CallbackManager;
import com.wan.net.rpc.invoker.callback.RpcFuture;
import com.wan.net.rpc.invoker.client.RpcClient;
import com.wan.net.rpc.message.RpcRequest;
import com.wan.net.rpc.message.RpcResponse;
import com.wan.net.rpc.registry.Registry;
import com.wan.net.rpc.service.RpcService;
import com.wan.net.rpc.util.RpcUtil;
import com.wan.thread.actor.Actor;
import com.wan.thread.actor.ActorExecutor;
import com.wan.util.ClassUtil;
import com.wan.util.SplitUtil;
import com.wan.util.number.ByteIncrement;
import com.wan.util.serializer.ProtoStuffSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * rpc消息调用器，通过构建invoker来定向发送消息
 * @author wwh
 * @date 2022/6/22 11:06
 */
public class RpcInvoker {
    private static final Logger logger = LoggerFactory.getLogger(RpcInvoker.class);
    private static final ByteIncrement BYTE_INCREMENT = new ByteIncrement((byte) 0);
    private static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger(0);
    private static final Object[] emptyArgs = new Object[0];
    private static final ActorExecutor EXECUTOR;
    static {
        EXECUTOR = new ActorExecutor(Runtime.getRuntime().availableProcessors() + 2, "rpc-invoke");
    }
    /** 需调用的服务 */
    private final RpcService rpcService;
    /** 当前发送器的临时id */
    private final byte tempId;
    /** 服务均衡选取器 */
    private final Balancer<RpcService> balancer;
    /** 访问服务的客户端 */
    private final Client client;
    /** 请求超时 */
    private final int timeout;
    /** 被调用的服务 */
    private RpcService serviceCaller;

    private RpcInvoker(RpcService rpcService,Balancer<RpcService> balancer,int timeout) {
        this.rpcService = rpcService;
        this.tempId = BYTE_INCREMENT.next();
        this.balancer = balancer;
        this.client = findClient();
        this.timeout = timeout;
    }

    public static <T>Builder<T> builder(Class<T> serviceInterface){
        Assert.isTrue(serviceInterface.isInterface(),"rpc服务必须是接口->{}",serviceInterface);
        return new Builder<>(serviceInterface);
    }

    private Client findClient(){
        Registry registry = RpcManager.getRegistry();
        List<RpcService> rpcServices = registry.find(rpcService);
        String serviceName = rpcService.toString();
        if(rpcServices == null || rpcServices.isEmpty()){
            throw new RpcInvokeException(serviceName,"不存在该服务");
        }
        serviceCaller = balancer.choose(rpcServices);
        if(serviceCaller == null){
            throw new RpcInvokeException(serviceName,"选择null服务->balancer:" + balancer.getClass());
        }
        if(serviceCaller.getHostAndPort() == null){
            throw new RpcInvokeException(serviceName,"选择的服务地址为空->service:" + serviceCaller);
        }
        Client client = Client.get(serviceCaller.getHostAndPort());
        if(client == null){
            client = new RpcClient(serviceCaller.getHostAndPort());
        }
        return client;
    }

    /**
     * rpc远程调用，根据返回值和回调方法决定是否阻塞等待结果返回
     * @param methodName 方法名
     * @param args 参数
     * @return 结果
     */
    public Object invoke(String methodName, Consumer<RpcResponse> callback,Object...args){
        try {
            int messageId = ATOMIC_INTEGER.incrementAndGet();
            final Object[] args_ = args == null ? emptyArgs : args;
            //方法签名验证
            boolean hasReturn; //是否有返回值
            Method method = null;
            try {
                for (Method method_ : rpcService.getServiceInterface().getMethods()) {
                    if(method_.getName().equals(methodName) &&
                            ClassUtil.checkObjectArrayEqualType(args_,method_.getParameterTypes())){
                        method = method_;
                        break;
                    }
                }
                if(method == null){
                    List<String> strings = Arrays.stream(args_).map(a -> a.getClass().getName()).collect(Collectors.toList());
                    throw new NoSuchMethodException(rpcService.getServiceInterface().getName() + "." + methodName +
                            "(" + SplitUtil.concat(strings) + ")");
                }
                hasReturn = !void.class.equals(method.getReturnType());
            } catch (Exception e) {
                throw new RpcInvokeException(rpcService.getServiceInterface().getCanonicalName(),"服务方法验证失败",e);
            }
            Object implementService = serviceCaller.getParams().get(RpcService.IMPLEMENT);
            String serviceMethodName = RpcUtil.buildServiceMethodName(rpcService.getServiceInterface(),
                    methodName,method.getParameterTypes(),implementService.toString());
            //先绑定消息future
            RpcFuture<RpcResponse> future = new RpcFuture<>(callback);
            CallbackManager.addFuture(messageId,future);
            //异步执行发送
            EXECUTOR.execute(new Actor() {
                @Override
                public long asyncId() {
                    return tempId;
                }

                @Override
                public void run() {
                    send(messageId,serviceMethodName,args_);
                }
            });
            //有返回值，或者回调函数为空则阻塞等待，有回调函数强制异步不等待
            if(hasReturn && callback == null){
                try {
                    RpcResponse response = future.get(timeout, TimeUnit.MILLISECONDS);
                    return response.getResult();
                }catch (Exception e){
                    throw new RpcInvokeException(serviceMethodName,"回调结果异常",e);
                }
            }
        }catch (Throwable e){
            logger.error("rpc远程调用失败");
            throw e;
        }
        return null;
    }

    public Object invoke(String methodName,Object...args){
        return this.invoke(methodName,null,args);
    }

    private void send(int messageId,String serviceMethodName,Object[] args){
        RpcRequest request = new RpcRequest();
        request.setTempId(tempId);
        request.setServiceName(serviceMethodName);
        request.setParams(args);
        Packet packet = new Packet();
        packet.setMessageId(messageId);
        try {
            byte[] encode = ProtoStuffSerializer.encode(request);
            packet.setContent(encode);
        } catch (IOException e) {
            throw new RpcInvokeException(serviceMethodName,"序列化出错",e);
        }
        client.send(packet);
    }

    /**
     * 获取代理服务类
     */
    private <T>T proxy(){
        Class<?> proxyInterface = rpcService.getServiceInterface();
        //noinspection unchecked
        return (T) Proxy.newProxyInstance(RpcInvoker.class.getClassLoader(), new Class[]{proxyInterface},
                (proxy, method, args) -> invoke(method.getName(),args));
    }

    public static class Builder<T> {
        private final Class<T> serviceInterface;
        private final Map<String,Object> params = new HashMap<>();
        private HostAndPort hostAndPort;
        //未设置时，默认使用随机策略
        private Balancer<RpcService> balancer = Balancer.Random;
        private int timeout = 4000;

        private Builder(Class<T> serviceInterface) {
            this.serviceInterface = serviceInterface;
        }

        public Builder<T> param(String key,Object value){
            this.params.put(key,value);
            return this;
        }
        /** 指定选择器 */
        public Builder<T> balancer(Balancer<RpcService> balancer){
            this.balancer = balancer;
            return this;
        }
        /** 设置请求超时时间 */
        public Builder<T> timeout(int timeout){
            this.timeout = timeout;
            return this;
        }
        /** 指定实现类 */
        public Builder<T> implement(String implementService){
            return this.param(RpcService.IMPLEMENT,implementService);
        }
        /** 指定版本号 */
        public Builder<T> version(String version){
            return this.param(RpcService.VERSION,version);
        }
        /** 指定目标地址 */
        public Builder<T> hostAndPort(HostAndPort hostAndPort){
            this.hostAndPort = hostAndPort;
            return this;
        }

        public T proxy(){
            RpcInvoker rpcInvoker = buildInvoker();
            return rpcInvoker.proxy();
        }

        public RpcInvoker buildInvoker(){
            RpcService rpcService = RpcService.create().serviceInterface(serviceInterface);
            rpcService.getParams().putAll(params);
            if(hostAndPort != null){
                rpcService.hostAndPort(hostAndPort);
            }
            return new RpcInvoker(rpcService,balancer,timeout);
        }
    }
}
